package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 生产者
 */
public class ProducerQuickStart {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1.设置配置参数
        Properties prop = new Properties();
        //kafka链接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //key和value的序列化
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //ack配置  消息确认机制
        //0 : 消息一旦发出则认为成功（消息有丢失风险，但速度最快）
        //1 : 只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应
        //all : 只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应
        prop.put(ProducerConfig.ACKS_CONFIG,"all");
        //重试次数
        prop.put(ProducerConfig.RETRIES_CONFIG,10);
        //数据压缩
        //snappy : 占用较少的 CPU，却能提供较好的性能和相当可观的压缩比，如果看重性能和网络带宽，建议采用
        //lz4 : 占用较少的 CPU，压缩和解压缩速度较快，压缩比也很客观
        //gzip : 占用较多的 CPU，但会提供更高的压缩比，网络带宽有限可以使用这种算法
        prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");

        //2.创建kafka生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<>(prop);

        //3.发送消息
        /**
         * 第一个参数 ：topic
         * 第二个参数：消息的key
         * 第三个参数：消息的value
         */
        //ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("topic-first","lxl","hello kafka");

        //测试kafkaStreams实时流计算
        ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("itcast-topic-input","hello kafka");
        for (int i = 0; i < 7; i++) {
            producer.send(kvProducerRecord);
        }

        //同步发送消息
        /*producer.send(kvProducerRecord);
        RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
        System.out.println(recordMetadata.offset());*/

        //异步消息发送
        /*producer.send(kvProducerRecord, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if (e != null){
                    e.printStackTrace();
                }
                System.out.println(recordMetadata.offset());
            }
        });*/

        //4.关闭消息通道  必须要关闭，否则消息发送不成功
        producer.close();
    }

}
